from pyspark.sql import SparkSession
import os

"""
-------------------------------------------------
   Description :	TODO：
   SourceFile  :	Demo03_WriteToES
   Author      :	itcast team
-------------------------------------------------
"""

# 0.设置系统环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python3'

# 构建SparkSession
# 建造者模式：类名.builder.配置…….getOrCreate()
# 自动帮你构建一个SparkSession对象，只要指定你需要哪些配置就可
# spark = SparkSession \
#     .builder \
#     .master("local[2]") \  # 本地模式运行，使用2个CPU核心
#     .appName("SparkSQLAppName") \  # 设置应用程序名称
#     .config("spark.sql.shuffle.partitions", 4) \  # 设置shuffle分区数
#     .config("hive.metastore.uris", "thrift://up01:9083") \  # 配置Hive元数据服务地址
#     .config("hive.metastore.warehouse.dir", "hdfs://up01:8020/user/hive/warehouse") \  # 配置Hive数据仓库路径
#     .enableHiveSupport() \  # 启用Hive支持
#     .getOrCreate()  # 获取或创建SparkSession实例
spark = SparkSession \
    .builder \
    .master("local[2]") \
    .appName("SparkSQLAppName") \
    .config("spark.sql.shuffle.partitions", 4) \
    .config("hive.metastore.uris", "thrift://up01:9083") \
    .config("hive.metastore.warehouse.dir", "hdfs://up01:8020/user/hive/warehouse") \
    .enableHiveSupport() \
    .getOrCreate()

#2.读取数据（从Hive）
#使用SparkSQL读取Hive中default数据库的tb_hdfs表数据
input_df = spark.read.table("default.tb_hdfs")

#3.数据处理
input_df.printSchema()
input_df.show()

#4.数据写出（ES）
# input_df.write\
#     .format("es") \  # 指定数据源格式为Elasticsearch
#     .option("es.resource", "test_spark_write_es") \  # 指定ES索引名称
#     .option("es.nodes", "up01:9200") \  # 指定ES节点地址和端口
#     .option("es.mapping.id", "id") \  # 指定文档ID对应的字段
#     .save()  # 执行保存操作
input_df.write\
    .format("es")\
    .option("es.resource","test_spark_write_es")\
    .option("es.nodes","up01:9200")\
    .option("es.mapping.id","id")\
    .save()

# 关闭SparkSession
spark.stop()
